package com.atguigu.flink.chapter11.function;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * @Author lzc
 * @Date 2022/11/1 14:51
 */
public class Flink02_Function_Table {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        DataStreamSource<WaterSensor> stream = env.fromElements(
            new WaterSensor("hello abc", 1L, 10),
            new WaterSensor("hello atguigu abc", 1L, 10),
            new WaterSensor("hello hello", 2L, 20)
        );
    
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        Table table = tEnv.fromDataStream(stream);
        tEnv.createTemporaryView("sensor", table);
        
        // 把小字符串变成大写的函数
        // 1. 在 table api 中使用
        // 1.1 内联的方式
       
        // 1.2 先注册再使用
        tEnv.createTemporaryFunction("my_split", MySplit.class);
       /* table
            // 内连接
//            .joinLateral(call("my_split", $("id"))) // 把一个 id 给且成的了多行多列
            .leftOuterJoinLateral(call("my_split", $("id"))) // 把一个 id 给且成的了多行多列
            .select($("id"), $("word"), $("len"))
            .execute()
            .print();
       */
        
        // 2. 在 sql 中
        // 先注册
        // select  form a join b on a.id=b.id
        // select  form a,b where a.id=b.id
        /*tEnv
            .sqlQuery("select " +
                          " id, w, l " +
                          "from sensor " +
                          "join lateral table( my_split(id) ) as t(w, l) on true")
            .execute()
            .print();*/
        // 内连接
        /*tEnv
            .sqlQuery("select " +
                          " id, word, len " +
                          "from sensor " +
                          ", lateral table( my_split(id) )")
            .execute()
            .print();*/
        
        tEnv
            .sqlQuery("select " +
                          " id, w, l " +
                          "from sensor " +
                          "left join lateral table( my_split(id) ) as t(w, l) on true")
            .execute()
            .print();
        
    
    }
    
    // 给每列定义列名和类型
    @FunctionHint(output = @DataTypeHint("row<word string, len int>"))
    public static class MySplit extends TableFunction<Row> {
        // 1. 返回值类型必须是 void
        // 2. 方法名必须是 eval
        // 3. 参数根据实际情况
        public void eval(String s){
            if (s.contains("atguigu")) {
                return;
            }
            // 这个方法调用几次,s 就生成几行
            for (String word : s.split(" ")) {
                collect(Row.of(word, word.length()));
            }
        }
    }
}
/*
制表函数

每给你一个字符串, 把字符串切割, 并得到切割后的字符串和它的长度
"hello abc"
    hello  5
    abc    3
"hello atguigu abc"
    hello  5
    atguigu 7
    abc  3


....

----
"hello abc"              hello  5
"hello abc"              abc    3
"hello atguigu abc"      hello  5
"hello atguigu abc"      atguigu 7
"hello atguigu abc"      abc  3

 */